package org.test.rocketmq.consumer02;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

/**
 * Pull消费:比较原始Pull Consumer
 */
public class PullConsumer {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("pull_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.start();

        MessageQueue mq = new MessageQueue();
        mq.setQueueId(0);
        mq.setTopic("TopicTest");
        mq.setBrokerName("81d041023b08");
        long offset = 26;
        PullResult pullResult = consumer.pull(mq, "*", offset, 1);
        if (pullResult.getPullStatus().equals(PullStatus.FOUND)) {
            System.out.printf("%s%n", pullResult.getMsgFoundList());
            consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
        }
        consumer.shutdown();
    }
}
